package com.imooc.producer.b_order.overall;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;

import static com.utils.Const.NAME_SERVER_SINGLE_LOCAL;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2024/4/9-2024
 * @Description: 全局有序性
 *      要想保证全局有序性，那么该 topic 在整个broker 集群中只能有一个队列。
 */
public class AllOrderlyProducer {
    public static void main(String[] args) throws Exception {
        //定义 producer
        DefaultMQProducer producer = new DefaultMQProducer("all_orderly");
        producer.setNamesrvAddr(NAME_SERVER_SINGLE_LOCAL);

        //设置 queue 的个数。
        //该参数的设置十分重要，因此要全局有序，因此，必须只有一个队列。所有的消息全部发送到一个队列中。
        producer.setDefaultTopicQueueNums(1);

        //启动 producer
        producer.start();

        //发送消息
        for (int i = 0; i < 10; i++) {
            Integer orderId = i;
            byte[] body = ("Hi, all orderly" + i).getBytes();
            Message message = new Message("all_orderly_topic", "tags_" + i, "key_" + i, body);
            producer.send(message, new MessageQueueSelector() {
                /**
                 * 保证消息的有序性，本质就是将需要保证顺序的这一批消息都发送到一个队列中。因为队列本身就是先进先出，队列是能够保证消息的顺序的。
                 * 因此，需要保证顺序的消息，发送到一个队列中。这样这些消息在消费的时候就是顺序的。
                 * 怎么保证同一批消息发送到一个队列中？
                 *      MessageQueueSelector  这个接口给我我们一个方案。
                 *      通过实现该接口，在该接口中可以根据我们自己的一个算法，指定消息发送哪个队列中。
                 *      比如：有一批消息需要保证顺序，这批消息中的 parentId 属性是一样的，  那么我们就可以使用属性进行算法的运算，计算出一个队列。
                 *      这批消息的 parentId 都是一样的，最终每个消息运算的结果都是一样的，这样这一批消息最终会选择到同一个队列进行发送。
                 *      这样就可以实现消息的顺序。
                 *  顺序消息出了生产者需要进行特殊的处理，
                 */
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Integer id = (Integer) arg;
                    int num = id % mqs.size();
                    return mqs.get(num);
                }
            }, orderId);
        }

        //关闭生产者
        producer.shutdown();
    }
}